package top.malaoshi.controller;

import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
public class KafkaPrdCtrl {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/send/{msg}")
    public String send(@PathVariable String msg){
        kafkaTemplate.send("userlog2", msg);//向notice主题发送消息

        String content=String.format("msg：%s", msg);
        System.out.println(content);
        return "消息已经发送";
    }

    @GetMapping("/send2/{msg}")
    public String send2(@PathVariable String msg){
        kafkaTemplate.send("userlog2", msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败：" + ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                String content = String.format("发送消息成功：%s - %s - %s",
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                System.out.println(content);
            }
        });//向notice主题发送消息

        String content=String.format("msg：%s", msg);
        System.out.println(content);
        return "消息已经发送";
    }

    @GetMapping("/send6/{msg}")
    public String send6(@PathVariable String msg){
        kafkaTemplate.send("userlog2","like", msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败：" + ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                String content = String.format("发送消息成功：%s - %s - %s",
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                System.out.println(content);
            }
        });//向notice主题发送消息

        String content=String.format("msg：%s", msg);
        System.out.println(content);
        return "消息已经发送";
    }

    @GetMapping("/send3/{msg}")
    public String send3(@PathVariable String msg){
        ListenableFuture<SendResult<String,Object>> future = kafkaTemplate.send("userlog2", msg);//向notice主题发送消息

        try {
            SendResult<String,Object> result = future.get();
            String content = String.format("发送消息成功：%s - %s - %s",
                    result.getRecordMetadata().topic(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            System.out.println(content);
        }catch (Throwable e){
            e.printStackTrace();
        }
        return "消息已经发送";
    }


//    @GetMapping("/send4/{msg}")
//    public String send4(@PathVariable("msg") String msg) {
//        // 声明事务：后面报错消息不会发出去
//        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback(){
//            @Override
//            public Object doInOperations(KafkaOperations operations) {
//                operations.send("userlog2", msg);
//                if("error".equals(msg)){
//                    int i=1/0;//此处报错，让上面send()方法回滚
//                }
//                return null;
//            }
//        });
//        return "消息已发送";
//    }
//
//
@GetMapping("/send5/{msg}")
@Transactional(rollbackFor = Exception.class)
public String send5(@PathVariable("msg") String msg){
    //事务的支持
    kafkaTemplate.send("userlog2",msg);
    if("error".equals(msg)){
        int i=1/0;
    }
    kafkaTemplate.send("userlog2",msg+"--2");
    return "消息已发送";
}
}
